package com.itsu.job.context;

import cn.hutool.extra.spring.SpringUtil;
import com.itsu.job.components.ScheduleHandler;
import com.itsu.job.components.ScheduleRunnable;
import com.itsu.job.config.Job;
import com.itsu.job.config.JobProperties;
import com.itsu.job.exception.JobException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.stream.Collectors;

/**
 * @author jerry
 * @Date 2022/2/7
 */
@Slf4j
public class JobHelper {

    public static final ConcurrentHashMap<String, ScheduledFuture<?>> scheduleFuture = new ConcurrentHashMap<>();

    final ThreadPoolTaskScheduler threadPoolTaskScheduler;

    final JobProperties jobProperties;

    public JobHelper(ThreadPoolTaskScheduler threadPoolTaskScheduler, JobProperties jobProperties) {
        this.threadPoolTaskScheduler = threadPoolTaskScheduler;
        this.jobProperties = jobProperties;
    }

    /**
     * schedule a job
     *
     * @param job
     */
    public void scheduleJob(Job job) {
        if (!StringUtils.hasText(job.getJobId()) || !StringUtils.hasText(job.getCron())) {
            throw new JobException("can not register a job without a jobID or cron");
        }
        if (!StringUtils.hasText(job.getHandlerName())) {
            log.warn("job id:{} does not have a handler, will use job id to handle", job.getJobId());
            job.setHandlerName(job.getJobId());
        }
        Object handler = SpringUtil.getBean(job.getHandlerName());
        if (handler == null) {
            throw new JobException("job id:" + job.getJobId() + ", job handler:" + job.getHandlerName()
                    + " is not in the Spring Application Context");
        }
        if (!(handler instanceof ScheduleHandler)) {
            throw new JobException(
                    "job id:" + job.getJobId() + ", job handler:" + handler + " is not instanceof ScheduleHandler");
        }
        ScheduleHandler jh = (ScheduleHandler) handler;
        ScheduledFuture<?> future = threadPoolTaskScheduler.schedule(new ScheduleRunnable(jh, job),
                triggerContext -> new CronTrigger(job.getCron()).nextExecutionTime(triggerContext));
        Assert.notNull(future, "schedule future is null");
        scheduleFuture.put(job.getJobId(), future);
        log.info("register job with information {}", job);
    }

    /**
     * stop the job, which is already scheduled
     *
     * @param jobId
     * @param type
     */
    public void stop(String jobId, Integer type) {
        if (scheduleFuture.containsKey(jobId)) {
            ScheduledFuture<?> future = scheduleFuture.remove(jobId);
            type = type == null ? 0 : 1;
            future.cancel(type == 0);
        } else {
            throw new JobException("current job: " + jobId + " is not running");
        }
    }

    /**
     * run the job once
     *
     * @param jobId
     */
    public void runJob(String jobId) {
        if (!StringUtils.hasText(jobId)) {
            throw new JobException("jobId not found");
        }
//        List<Job> jobs = jobProperties.getJobs().stream().filter(x -> x.getJobId().equals(jobId))
//                .collect(Collectors.toList());
//        if (CollUtil.isEmpty(jobs)) {
//            throw new JobException("config job is empty");
//        }
//        Job job = jobs.get(0);
        Job job = getJobById(jobId);
        String handlerName = job.getHandlerName();
        if (!StringUtils.hasText(handlerName)) {
            handlerName = job.getJobId();
        }
        Object handler = SpringUtil.getBean(handlerName);
        if (!(handler instanceof ScheduleHandler)) {
            throw new JobException("handler must be implements " + ScheduleHandler.class.getName());
        }
        ScheduleHandler sh = (ScheduleHandler) handler;
        sh.handle(job);
    }

    /**
     * 1 for scheduled 0 for not scheduled
     *
     * @param jobId
     * @return
     */
    public int status(String jobId) {
        return scheduleFuture.containsKey(jobId) ? 1 : 0;
    }

    public Map<String, Integer> statusAll(String jobIdStr) {
        if (StringUtils.hasText(jobIdStr)) {
            String[] jobIds = jobIdStr.split(",");
            Map<String, Integer> map = new LinkedHashMap<>();
            for (String jobId : jobIds) {
                map.put(jobId, scheduleFuture.containsKey(jobId) ? 1 : 0);
            }
            return map;
        } else {
            return null;
        }
    }

    /**
     * judge if the jobId is scheduled
     *
     * @param jobId
     * @return
     */
    public boolean isStarted(String jobId) {
        return JobHelper.scheduleFuture.containsKey(jobId);
    }

    /**
     * stop the job which is scheduled earlier
     *
     * @param jobId
     */
    public void stop(String jobId) {
        stop(jobId, 1);
    }

    /**
     * judge if the job is schedule in local machine
     *
     * @param jobId
     * @return
     */
    public boolean isLocalJob(String jobId) {
        return jobProperties.getJobs().stream().map(Job::getJobId).collect(Collectors.toList()).contains(jobId);
    }

    /**
     * GET the job by jobId
     *
     * @param jobId
     * @return
     */
    public Job getJobById(String jobId) {
        List<Job> jobs = jobProperties.getJobs().stream().parallel().filter(x -> jobId.equals(x.getJobId())).collect(Collectors.toList());
        Assert.notEmpty(jobs, "jobId: " + jobId + " is not found in config");
        return jobs.get(0);
    }
}
